package com.pocket.insight.base.mql;

import com.pocket.insight.entity.ME;
import com.pocket.insight.entity.Type;
import lombok.extern.slf4j.Slf4j;
import org.jparsec.*;
import org.jparsec.pattern.Patterns;
import org.yaml.snakeyaml.parser.ParserException;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.function.BinaryOperator;
import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 描述：
 *
 * @author shuize
 * @create 2024/12/9
 */
@Slf4j
public class MQL {

    private static final String[]            OPERATORS            = new String[]{"+", "-", "*", "/", "%", ">", "<", "=", ">=", "<=", "<>", "!=", ".", ",", "(", ")", "[", "]", "^", "{", "}", "$"};
    private static final String[]            KEYWORDS             = new String[]{"select", "from", "as", "where", "group", "by", "having", "order", "asc", "desc", "and", "or", "not", "in", "exists", "between", "is", "null", "like", "escape", "inner", "outer", "left", "right", "full", "cross", "join", "on", "union", "all", "case", "when", "then", "else", "end", "true", "false", "over", "window", "partition", "rows", "preceding", "following", "unbounded", "current", "row", "filter", "within", "date", "timestamp", "nulls", "first", "last"};
    private static final Parser<ME>          BRACKET_ATTR         = Patterns.notString("}}").many().toScanner("bracket quote").between(Scanners.string("{{$"), Scanners.string("}}")).source().map(
            s -> ME.create(ME.Kind.attr, s.substring("{{$".length(), s.length() - "}}".length()))
    );
    private static final Parser<ME>          BRACKET_PARAM        = Patterns.notString("}}").many().toScanner("bracket quote").between(Scanners.string("{{%"), Scanners.string("}}")).source().map(
            s -> ME.create(ME.Kind.param, s.substring("{{%".length(), s.length() - "}}".length()))
    );
    private static final Parser<FieldNode>   BRACKET_QUOTE_STRING = Patterns.notString("}").many().toScanner("bracket quote").between(Scanners.isChar('{'), Scanners.isChar('}')).source().map(
            s -> new FieldNode(s.substring(1, s.length() - 1))
    );
    private static final Parser<DatasetNode> BRACKET_DATASET      = Patterns.notString("}}").many().toScanner("bracket quote").between(Scanners.string("{{"), Scanners.string("}}")).source().map(
            s -> new DatasetNode(s.substring("{{".length(), s.length() - "}}".length()))
    );

    private static final Terminals            TERMS;
    private static final Parser<?>            TOKENIZER;
    private static final Parser<ME>           literalNull;
    private static final Parser<ME>           literalBool;
    private static final Parser<ME>           literalNumber;
    private static final Parser<ME>           literalString;
    private static final Parser<ME>           literalDate;
    private static final Parser<ME>           literalTime;
    private static final Parser<ME>           simpleHE;
    private static final Parser<ME>           dataset;
    private static final Parser<ME>           simpleIdentifier;
    private static final Parser<ME>           compoundIdentifier;
    private static final Parser.Reference<ME> expressionRef;
    private static final Parser<ME>           filterExpression;
    private static final Parser<ME>           fieldFilter;
    private static final Parser<ME>           identifier;
    private static final Parser<ME>           expressionWithDirection;
    private static final Parser<List<ME>>     withinGroup;
    private static final Parser<Integer>      windowFrame;
    private static final Parser<ME.Window>    windowDefination;
    private static final Parser<ME>           parentExpression;
    private static final Parser<String>       rawIdendifier;
    private static final Parser<ME>           functionCall;
    private static final Parser<ME>           caseExpression;
    private static final Parser<ME>           array;
    private static final Parser<ME>           arithmeticOperand;
    private static final Parser<ME>           arithmetic;
    private static final Parser<ME>           expressionWithoutAs;
    private static final Parser<ME>           asExpression;


    static {
        TERMS = Terminals.operators(OPERATORS).words(Scanners.IDENTIFIER).caseInsensitiveKeywords(Arrays.asList(KEYWORDS)).build();
        TOKENIZER = Parsers.or(Terminals.DecimalLiteral.TOKENIZER, Terminals.StringLiteral.SINGLE_QUOTE_TOKENIZER, BRACKET_ATTR, BRACKET_PARAM, BRACKET_DATASET, BRACKET_QUOTE_STRING, TERMS.tokenizer());
        literalNull = term("null").map(
                s -> ME.create(ME.Kind.constant, null)
        );
        literalBool = Parsers.or(term("true"), term("false")).map(
                s -> createConstant(Type.TypeName.bool, s.toString())
        );
        literalNumber = Terminals.DecimalLiteral.PARSER.map(
                s -> createConstant(Type.TypeName.number, s)
        );
        literalString = Terminals.StringLiteral.PARSER.map(
                s -> createConstant(Type.TypeName.string, s)
        );
        literalDate = term("date").next(Terminals.StringLiteral.PARSER).map(
                s -> createConstant(Type.TypeName.date, s)
        );
        literalTime = term("timestamp").next(Terminals.StringLiteral.PARSER).map(
                s -> createConstant(Type.TypeName.time, s)
        );
        simpleHE = Parsers.tokenType(ME.class, "bracket quote").map(
                f -> f
        );
        dataset = Parsers.tokenType(DatasetNode.class, "dataset").map(
                d -> ME.create(ME.Kind.dataset, d.dataset)
        );
        simpleIdentifier = Parsers.tokenType(FieldNode.class, "bracket quote").map(
                f -> ME.create(ME.Kind.field, f.fieldName)
        );
        compoundIdentifier = Parsers.sequence(simpleIdentifier.or(dataset), term("."), simpleIdentifier,
                (d, _ignore, f) -> ME.create(ME.Kind.field, f.op).setDataset(d.op)
        );
        expressionRef = Parser.newReference();
        filterExpression = term("filter").next(paren(term("where").next(expressionRef.lazy())));
        fieldFilter = Parsers.sequence(Parsers.or(compoundIdentifier, simpleIdentifier), filterExpression.optional(null), (expr, filter) -> {
            if (filter != null) {
                expr.filter = new ArrayList(Arrays.asList(filter));
            }
            return expr;
        });
        identifier = Parsers.or(fieldFilter, compoundIdentifier, dataset, simpleHE, simpleIdentifier);
        expressionWithDirection = Parsers.sequence(expressionRef.lazy(), Parsers.or(term("asc"), term("desc")).optional(null), Parsers.sequence(term("nulls"), Parsers.or(term("first"), term("last")),
                (s1, direction) -> direction.toString()
        ).optional(null), (expr, direction, nullsOrder) -> {
            if (direction != null) {
                expr.direction = direction.toString();
            }
            if (nullsOrder != null) {
                expr.nullsOrder = ME.NullsOrder.LAST.toString().equalsIgnoreCase(nullsOrder) ? ME.NullsOrder.LAST : ME.NullsOrder.FIRST;
            }
            return expr;
        });
        withinGroup = term("within").next(term("group")).next(paren(Parsers.sequence(term("order"), term("by"), expressionWithDirection.sepBy1(term(",")))));
        windowFrame = Parsers.or(Parsers.sequence(Terminals.DecimalLiteral.PARSER.map(
                s -> Integer.parseInt(s)
        ).or(term("unbounded").map(
                s -> null
        )), Parsers.or(term("following"), term("preceding")), (offset, followingOrPreceding) -> {
            if (offset == null) {
                return null;
            } else {
                if ("preceding".equalsIgnoreCase(followingOrPreceding.toString())) {
                    offset = -offset;
                }
                return offset;
            }
        }), term("current").next(term("row")).map(
                s -> 0
        ));
        windowDefination = term("over").next(paren(Parsers.sequence(term("partition").next(term("by")).next(expressionRef.lazy().sepBy1(term(","))).optional(null), term("order").next(term("by")).next(expressionWithDirection.sepBy1(term(","))).optional(null), term("rows").next(term("between")).next(Parsers.sequence(windowFrame, term("and"), windowFrame,
                (start, _ignored, end) -> {
                    ME.Window w = new ME.Window();
                    w.lowerBound = start;
                    w.upperBound = end;
                    w.hasFrameClause = true;
                    return w;
                })).optional(null), (partition, order, framingWindow) -> {
            ME.Window w = new ME.Window();
            w.partition = partition;
            w.order = order;
            if (framingWindow != null) {
                w.lowerBound = framingWindow.lowerBound;
                w.upperBound = framingWindow.upperBound;
                w.hasFrameClause = framingWindow.hasFrameClause;
            }
            return w;
        })));
        parentExpression = paren(expressionRef.lazy());
        rawIdendifier = term("$").next(term(".")).next(Terminals.identifier()).map(
                s -> "$." + s
        );
        functionCall = Parsers.sequence(Terminals.identifier().or(rawIdendifier).or(Terminals.RESERVED), paren(expressionRef.lazy().sepBy(term(","))), filterExpression.optional(null), withinGroup.optional(null), windowDefination.optional(null), (op, args, f, within, window) -> {
            ME expr = ME.create(ME.Kind.function, op, Collections.singletonList(args));
            expr.window = window;
            expr.within = within;
            if (f != null) {
                expr.filter = new ArrayList(Arrays.asList(f));
            }

            return expr;
        });
        caseExpression = Parsers.sequence(term("case"), Parsers.sequence(term("when"), expressionRef.lazy(), term("then"), expressionRef.lazy(), (_1, cond, _2, val) -> {
            return Stream.of(val, cond);
        }).many1(), term("else").next(expressionRef.lazy()).optional(null), term("end"), (_1, cases, defaultValue, _4) -> {
            List<ME> op = (List) cases.stream().flatMap((e) -> {
                return e;
            }).collect(Collectors.toList());
            if (defaultValue != null) {
                op.add(defaultValue);
            }

            return ME.create(ME.Kind.casewhen, op);
        });
        array = squareBracket(expressionRef.lazy().sepBy(term(","))).map((es) -> {
            return ME.create(ME.Kind.array, es);
        });
        arithmeticOperand = Parsers.or(new Parser[]{array, parentExpression, functionCall, literalNumber, literalString, literalBool, literalNull, literalDate, literalTime, identifier, caseExpression});
        arithmetic = (new OperatorTable())
                .infixl(binary("+"), MDF.getOperatorPrecedence("+"))
                .infixl(binary("-"), MDF.getOperatorPrecedence("-"))
                .infixl(binary("*"), MDF.getOperatorPrecedence("*"))
                .infixl(binary("/"), MDF.getOperatorPrecedence("/"))
                .infixl(binary("%"), MDF.getOperatorPrecedence("%"))
                .infixl(binary("^"), MDF.getOperatorPrecedence("^"))
                .prefix(unary("-"), 6)
                .prefix(unary("+"), 6)
                .infixl(binary(">"), MDF.getOperatorPrecedence(">"))
                .infixl(binary(">="), MDF.getOperatorPrecedence(">="))
                .infixl(binary("<"), MDF.getOperatorPrecedence("<"))
                .infixl(binary("<="), MDF.getOperatorPrecedence("<="))
                .infixl(binary("="), MDF.getOperatorPrecedence("="))
                .infixl(binary("<>"), MDF.getOperatorPrecedence("<>"))
                .infixl(binary("!="), MDF.getOperatorPrecedence("!="))
                .infixl(binary("and"), MDF.getOperatorPrecedence("and"))
                .infixl(binary("or"), MDF.getOperatorPrecedence("or"))
                .prefix(unary("not"), MDF.getOperatorPrecedence("not"))
                .build(arithmeticOperand);
        expressionWithoutAs = Parsers.or(new Parser[]{arithmetic, caseExpression, literalNumber, parentExpression, functionCall, literalNull, literalDate, literalTime, identifier, literalString});
        asExpression = Parsers.sequence(expressionWithoutAs, term("as").next(Parsers.or(new Parser[]{Terminals.Identifier.PARSER.or(simpleIdentifier.map((e) -> {
            return (String) e.op;
        }))})).optional((Object) null), (expr, uid) -> {
            if (uid != null) {
                expr.uid = (String) uid;
            }

            return expr;
        });
        expressionRef.set(asExpression);
    }

    public static ME createConstant(Type.TypeName type, String s) {
        if (s == null) {
            return ME.create(ME.Kind.constant, (Object) null);
        } else {
            Object v;
            if (type == Type.TypeName.string) {
                v = s;
            } else {
                v = type.type().cast(s);
            }

            if (v instanceof BigDecimal && ((BigDecimal) v).scale() == 0) {
                v = ((BigDecimal) v).longValue();
            }

            return ME.create(ME.Kind.constant, v);
        }
    }

    private static Parser<BinaryOperator<ME>> binary(String op) {
        return term(op).retn((l, r) -> ME.create(ME.Kind.function, op, Arrays.asList(new Object[]{l, r})));
    }

    public static Parser<?> term(String term) {
        return TERMS.token(term);
    }

    private static <T> Parser<T> paren(Parser<T> parser) {
        return parser.between(term("("), term(")"));
    }

    private static <T> Parser<T> squareBracket(Parser<T> parser) {
        return parser.between(term("["), term("]"));
    }

    public ME parse(String formula) {
        try {
            return parse(asExpression, formula);
        } catch (ParserException e) {
            return null;
        }
    }

    public static <T> T parse(Parser<T> parser, String source) {
        return parser.from(TOKENIZER, Scanners.SQL_DELIMITER).parse(source);
    }

    private static Parser<UnaryOperator<ME>> unary(String op) {
        return term(op).retn((e) -> {
            ME newe = ME.from(e);
            if ("-".equals(op) && newe.kind == ME.Kind.constant) {
                if (newe.op instanceof Long) {
                    return ME.create(ME.Kind.constant, -(Long) newe.op);
                }

                if (newe.op instanceof BigDecimal) {
                    return ME.create(ME.Kind.constant, ((BigDecimal) newe.op).negate());
                }
            }

            return ME.create(ME.Kind.function, op, Arrays.asList(e));
        });
    }


    private static class FieldNode {
        public String fieldName;

        public FieldNode(String fieldName) {
            this.fieldName = fieldName;
        }
    }

    private static class DatasetNode {
        public Object dataset;

        public DatasetNode(Object dataset) {
            Long id = ME.parseDatasetId(dataset);
            if (id != null) {
                this.dataset = id;
            } else {
                this.dataset = dataset;
            }

        }
    }
}
